package com.chixing.controller;

import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
import com.chixing.service.IMessageService;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.websocket.*;
import jakarta.websocket.server.PathParam;
import jakarta.websocket.server.ServerEndpoint;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * WebSocket endpoint for push notifications.
 */
@Slf4j
@Component
@ServerEndpoint(value = "/testWebSocket/{id}")
public class WebSocketProcess {

    // 保持在线连接：userId -> WebSocketProcess 实例（含 Session）
    private static final ConcurrentHashMap<Long, WebSocketProcess> concurrentMap = new ConcurrentHashMap<>();

    // 会话对象（实例字段）
    private Session session;

    // RedisTemplate 注入到静态字段，供静态/实例方法使用
    private static RedisTemplate<String, Object> staticRedisTemplate;

    @Autowired
    public void setRedisTemplate(RedisTemplate<String, Object> redisTemplate) {
        WebSocketProcess.staticRedisTemplate = redisTemplate;
    }

    // IMessageService 同样注入到静态字段，以便在 ServerEndpoint 中使用
    private static IMessageService staticMessageService;

    @Autowired
    public void setMessageService(IMessageService messageService) {
        WebSocketProcess.staticMessageService = messageService;
    }

    private static final String ONLINE_USER_KEY = "online_user_key";
    private static final String OFFLINE_USER_KEY_PREFIX = "offline_user_key_";

    /**
     * 客户端建立连接时触发
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("id") Long userId) {
        this.session = session;
        concurrentMap.put(userId, this);

        try {
            if (staticRedisTemplate != null) {
                staticRedisTemplate.opsForSet().add(ONLINE_USER_KEY, userId);
            }
        } catch (Exception e) {
            log.warn("redis add online user fail", e);
        }

        log.info("WebSocket onOpen: userId={} connected", userId);

        // 1) 补发 Redis 中的离线消息（如果有）
        try {
            if (staticRedisTemplate != null) {
                String offlineKey = OFFLINE_USER_KEY_PREFIX + userId;
                if (staticRedisTemplate.hasKey(offlineKey)) {

                    // ✅ 从 Redis 拿的必须是 Object → String
                    List<Object> msgList = staticRedisTemplate.opsForList().range(offlineKey, 0, -1);

                    if (msgList != null && !msgList.isEmpty()) {
                        log.info("Found {} offline messages in Redis for user {}", msgList.size(), userId);

                        for (Object obj : msgList) {
                            String msg = String.valueOf(obj); // ✅ 转成 String
                            try {
                                session.getBasicRemote().sendText(msg);
                            } catch (IOException ioe) {
                                log.error("Failed to send offline redis msg to user {}", userId, ioe);
                            }
                        }

                        // ✅ 删除 Redis 离线消息
                        staticRedisTemplate.delete(offlineKey);
                    }
                }
            }
        } catch (Exception e) {
            log.error("Error while sending redis offline messages to user " + userId, e);
        }


        // 2) 补发数据库中未读消息，并标记为已读
        try {
            if (staticMessageService != null) {
                List<String> dbMsgs = staticMessageService.getUnreadMessages(userId); // ✅ 从 DB 拿未读消息

                if (dbMsgs != null && !dbMsgs.isEmpty()) {
                    for (String msg : dbMsgs) {
                        session.getBasicRemote().sendText(msg);
                    }
                    staticMessageService.markMessagesAsRead(userId); // ✅ 标记为已读
                }
            }
        } catch (Exception e) {
            log.error("Error while sending DB unread messages to user " + userId, e);
        }

    }

    /**
     * 接收到客户端消息
     */
    @OnMessage
    public void onMessage(String message, @PathParam("id") Long userId) {
        log.info("WebSocket received message from user {} : {}", userId, message);
        // 这里可以处理客户端发送的心跳或特定请求
    }

    /**
     * 连接关闭
     */
    @OnClose
    public void onClose(Session session, @PathParam("id") Long userId) {
        concurrentMap.remove(userId);
        try {
            if (staticRedisTemplate != null) {
                staticRedisTemplate.opsForSet().remove(ONLINE_USER_KEY, userId);
            }
        } catch (Exception e) {
            log.warn("redis remove online user fail", e);
        }
        log.info("WebSocket onClose: userId={} disconnected", userId);
    }

    /**
     * 出错处理
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("WebSocket error, session=" + session, error);
    }

    /**
     * 服务器向指定用户发送文本消息（同步发送）
     *
     * @param userId  接收者 userId
     * @param message 文本消息（必须为 JSON 字符串或文本）
     * @return true: 已在线并成功发送； false: 不在线（消息已保存到 redis + db）
     */
    public boolean sendMessage(Long userId, String message) {
        WebSocketProcess target = concurrentMap.get(userId);
        if (target != null && target.session != null && target.session.isOpen()) {
            try {
                target.session.getBasicRemote().sendText(message);
                return true;
            } catch (IOException e) {
                log.error("Send fail to online user {}", userId, e);
            }
        }

        // 用户不在线 → 存 Redis + DB（转 Integer）
        try {
            if (staticRedisTemplate != null) {
                staticRedisTemplate.opsForList().leftPush(OFFLINE_USER_KEY_PREFIX + userId, message);
            }
            if (staticMessageService != null) {
                staticMessageService.saveOfflineMessage(Math.toIntExact(userId), message); // ✅ 改这里
            }
            log.info("User {} offline, message stored", userId);
        } catch (Exception e) {
            log.error("Offline message save failed for user " + userId, e);
        }
        return false;
    }


    /**
     * 向所有在线客户端广播消息
     */
    public void sendAllMessage(String msg) {
        log.info("Broadcasting to {} clients", concurrentMap.size());
        Set<Map.Entry<Long, WebSocketProcess>> entries = concurrentMap.entrySet();
        for (Map.Entry<Long, WebSocketProcess> entry : entries) {
            WebSocketProcess wp = entry.getValue();
            try {
                if (wp != null && wp.session != null && wp.session.isOpen()) {
                    wp.session.getBasicRemote().sendText(msg);
                }
            } catch (IOException e) {
                log.warn("Broadcast send failed for user {}", entry.getKey(), e);
            }
        }
    }
}
